import concurrent
import json
import uuid
from datetime import timezone

from dateutil import parser
from kafka import KafkaProducer
from python_depend.gms_api_client import get_script_indicator
from python_depend.redis_pool import RedisPoolManager
from python_depend.time_urtils import  get_current_time, get_market_status, ge_time

# ==============================
# 初始化参数
# ==============================
# 最大线程数，ETF和股票分开实际执行X2
max_workers = 5
# 数据缓存key
redis_key_stock = "ds:magical_waves:stock_market_cap"
redis_key_etf = "ds:magical_waves:etf_assets"
# 盘后是否推送标记位
redis_key_after_push_flag = "ds:magical_waves:after_push_flag:"
# ==============================
# kafka初始化
# ==============================
#创建生产者实例
producer = KafkaProducer(
    bootstrap_servers=['extquote-kafka-common-kafka.hxkafka:9092'],  # Kafka 服务器地址
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),  # 自动序列化 JSON
    security_protocol="SASL_PLAINTEXT",
    sasl_mechanism="SCRAM-SHA-256",
    sasl_plain_username="Extquote",
    sasl_plain_password="Extquote@2024",
    api_version=(2, 5),
    request_timeout_ms=10000,
    max_block_ms=10000
)
# 创建测试环境
# bootstrap_servers = ['10.10.80.213:9092'],  # Kafka 服务器地址
# producer = KafkaProducer(
#     bootstrap_servers=['extquote-kafka-common-kafka-1:55587',
#                        'extquote-kafka-common-kafka-2:26212',
#                        'extquote-kafka-common-kafka-0:43938'],  # Kafka 服务器地址
#     value_serializer=lambda v: json.dumps(v).encode('utf-8'),  # 自动序列化 JSON
#     security_protocol="SASL_PLAINTEXT",
#     sasl_mechanism="SCRAM-SHA-256",
#     sasl_plain_username="Extquote",
#     sasl_plain_password="Extquote@2024",
#     api_version=(2, 5),
#     request_timeout_ms=10000,
#     max_block_ms=10000
# )
topic = 'extquote-screener-data'

# ==============================
# redis初始化
# ==============================
RedisPoolManager.initialize()
redis_client = RedisPoolManager.get('default')


# ==============================
# 调用gms解析
# ==============================
def get_magical_waves(code, market):
    offset = 2
    # 盘后取最后一根
    if ge_time(16, 0):
        # 比较时间，找出最小值
        offset = 1
    # 神奇电波信号参数
    params = {
        "code_list": {
            "market": market,
            "codes": [
                code
            ]
        },
        "trade_class": "TRADE_INTRADAY",
        "adjust_type": "FORWARD",
        "time_period": {
            "time_type": "MIN",
            "time_step": 1
        },
        "time_range": {
            "trade_date": -1,
            "offset_to_time": {
                "begin_offset": offset,
                "end_time": 0
            }
        },
        "calc_fields": [
            {
                "data_field": 4294967295
            }
        ]
    }
    try:
        data = get_script_indicator(params)
        cell = get_earliest_cell(data)
        return cell
    except Exception as e:
        print(f"[ERROR] 获取{market}/{code} 神奇电波数据失败: {e}")
        return None


# 获取满足条件的cell数据
def get_earliest_cell(data):
    if data is None:
        return ""
    min_time = None
    earliest_element = None
    cells_data_list = data.get('data', {}).get('data', [{}])[0].get('field_values', [{}])[0].get('cells', [{}])

    for cells_data in cells_data_list:
        time = cells_data["time"]
        dt = parser.isoparse(time)
        timestamp_ms = int(dt.astimezone(timezone.utc).timestamp() * 1000)
        # 如果当前大于UTC20点,则为盘后，取最大的那根
        # 比较时间，找出最小值
        if min_time is None or timestamp_ms < min_time:
            min_time = timestamp_ms
            earliest_element = cells_data
    return earliest_element


def push_to_kafka(market, code, data):
    push_info = {
        "id": "2134792347823",
        "type": "index",
        "data": {
            "index_id": "ext_mover_magic_signal_buy",
            "time": 1745374094085,
            "trade_date": 20250203,
            "trade_class": "intraday",
            "market": "186",
            "code": "SUNE"
        }
    }
    if data is None:
        return ""
    push_info.get("data")["code"] = code
    push_info.get("data")["market"] = market

    push_info.get("data")["trade_date"] = data["trade_date"]
    time = data["time"]
    dt = parser.isoparse(time)
    timestamp_ms = int(dt.astimezone(timezone.utc).timestamp() * 1000)
    push_info.get("data")["time"] = timestamp_ms
    plot_values = data.get('plot_value', [{}])

    for value_datum in plot_values:
        if "plot_shape" not in value_datum:
            continue
        plot_shape = value_datum.get("plot_shape")
        value = plot_shape.get("value")
        print(f"{code},value is : {str(value)}")
        if value in {1, 2, 3}:
            push_info.get("data")["index_id"] = "ext_mover_magic_signal_buy"
        elif value in {4}:
            push_info.get("data")["index_id"] = "ext_mover_magic_signal_sell"
        else:
            continue
        push_info["id"] = str(uuid.uuid1())
        # kafka异步消息推送
        try:
            # 盘后数据判断是否被推送过，防止半日市这种休市导致盘后一直执行。
            if trade_period == "post_market" or trade_period == "closed":
                push_flag_redis_key = redis_key_after_push_flag + code + str(timestamp_ms)
                push_flag = redis_client.get(push_flag_redis_key)
                if push_flag is None:
                    message_push(code, push_info)
                    print(f"盘后,进行推送缓存标记设置,key is {push_flag_redis_key}")
                    redis_client.set(push_flag_redis_key, "1", ex=86400)
                else:
                    print(f"{code},{str(timestamp_ms)}数据已经推送")
            else:
                message_push(code, push_info)
        except Exception as e:
            print(f"push message {str(push_info)} error: {e}")
            return None


def message_push(code, push_info):
    print(f"code:{code} push info is {str(push_info)}")
    producer.send(topic, push_info)
    print(f"{code} push info success")


# ==============================
# 获取标普500标的列表
# ==============================

def fetch_symbol_list():
    stock_list_str = redis_client.get(redis_key_stock)
    stock_list = json.loads(stock_list_str) if stock_list_str else []

    etf_list_str = redis_client.get(redis_key_etf)
    etf_list = json.loads(etf_list_str) if etf_list_str else []

    return stock_list, etf_list


def push_task(market, code):
    # 获取K线数据
    print(f"开始计算{code}数据")
    data = get_magical_waves(market=market, code=code)
    if data is None:
        print(f" {code}  data is empty")
        return
    push_to_kafka(market, code, data)


def submit_task_execute(symbol_list):
    if not symbol_list:
        print("[ERROR] 股票列表为空，任务终止")
        return
    print("查询标的数量：" + str(len(symbol_list)))
    # 线程池执行
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        # 2. 处理每只股票
        for stock in symbol_list:
            market = stock['market']
            code = stock['code']
            executor.submit(push_task, market, code)
    producer.flush()
    print("[COMPLETED] 消息发送完毕")


# ==============================
# 主执行函数
# ==============================
def execute_job():
    """主任务执行流程"""
    push_type = "${push_type}"
    # push_type = "STOCK"
    stock_list, etf_list = fetch_symbol_list()
    if push_type == "STOCK":
        print("提交股票计算任务")
        submit_task_execute(stock_list)
        print("所有任务提交完成,主线程继续执行")

    if push_type == "ETF":
        print("提交ETF计算任务")
        submit_task_execute(etf_list)
        print("[START] 开始执行任务")


def execute_check():
    # 盘前执行校验
    if trade_period == "intraday":
        if ge_time(9, 31):
            return True
        else:
            print(f"任务执行不在执行时间范围内")
            return False
    # 盘前执行校验，只需执行16:00分，用来补数据
    if trade_period == "post_market":
        current_minute = current_time.minute
        # 判断分钟数是否为 0
        if current_minute == 0:
            return True
        else:
            print(f"当前市场状态{trade_period},不满足分钟为0的执行条件")
            return False
    else:
        print(f"当前市场状态{trade_period}不满足执行条件")
        return False


if __name__ == '__main__':
    current_time = get_current_time()
    print(f"当前时间：{current_time}")
    # 获取市场状态
    market_status = get_market_status()
    if market_status is None:
        exit()
    market_status_data = market_status.get("data")
    print(f"当前市场状态：{market_status_data}")
    # 如果是在交易时段或者盘后则执行任务:
    trade_period = market_status_data["trade_period"]
    if execute_check():
        print("开始执行任务")
        execute_job()